package com.fishleap.client;

import com.fishleap.codec.NettyDecoder;
import com.fishleap.codec.NettyEncoder;
import com.fishleap.dto.RpcRequest;
import com.fishleap.dto.RpcResponse;
import com.fishleap.handler.UserClientHandler;
import com.fishleap.serialize.kyro.KryoSerializer;
import com.fishleap.util.CuratorUtils;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.api.CuratorWatcher;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.data.Stat;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author zlg
 * 消费者
 */
public class RPCConsumer {

    //1.创建一个线程池对象  -- 它要处理我们自定义事件
    private static ExecutorService executorService =
        Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    //2.声明一个自定义事件处理器  UserClientHandlerList
    // 不能直接在这里分配空间,new Random().nextInt(int i)会出现非法参数异常,参数为0的情况
    private static Map<String, UserClientHandler> userClientHandlerMap;

    // 获取zookeeper客户端
    private static CuratorFramework zkClient = CuratorUtils.zkClient;

    //3.编写方法,初始化客户端  ( 创建连接池  bootStrap  设置bootstrap  连接服务器)
    private static void initClient(RpcRequest rpcRequest) throws Exception {
        // 子节点列表发送变化后都将事件处理器列表清空
        userClientHandlerMap = new HashMap<>();
//        userClientHandlerList.clear();

        // 获取子节点列表并进行该节点监听
        String servicePath = "/"+rpcRequest.getInterfaceName();
        List<String> list = zkClient.getChildren().usingWatcher(new CuratorWatcher() {
            @Override
            public void process(WatchedEvent watchedEvent) throws Exception {
                // 进行节点列表监听
                if (watchedEvent.getType() == Watcher.Event.EventType.NodeChildrenChanged) {
                    initClient(rpcRequest);
                }
            }
        }).forPath(servicePath);
        System.out.println("当前 "+"/"+CuratorUtils.NAMESPACE+servicePath+" 节点的子节点列表为: ======>>>");
        list.forEach(System.out::println);

        // 通过zookeeper获取所有在线的远程服务
//        ServiceDiscovery serviceDiscovery = new ZkServiceDiscovery();
//        List<InetSocketAddress> list = serviceDiscovery.lookupServiceAll(rpcRequest.getInterfaceName());

        list.forEach(server -> {
            //1. 初始化UserClientHandler
            UserClientHandler userClientHandler  = new UserClientHandler();

            NioEventLoopGroup group = new NioEventLoopGroup();
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY,true)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline pipeline = socketChannel.pipeline();
//                        pipeline.addLast(new StringEncoder());
//                        pipeline.addLast(new StringDecoder());
                            pipeline.addLast(new NettyDecoder(new KryoSerializer(), RpcResponse.class));
                            pipeline.addLast(new NettyEncoder(new KryoSerializer(), RpcRequest.class));
                            pipeline.addLast(userClientHandler);
                        }
                    });

            //5.连接服务端  eg:127.0.0.1:8888
            String[] split = server.split(":");
            try {
                bootstrap.connect(split[0], Integer.parseInt(split[1])).sync();
            } catch (InterruptedException e) {
                e.getMessage();
            }
            userClientHandlerMap.put(server, userClientHandler);
        });

    }

    //4.编写一个方法,使用JDK的动态代理创建对象
    // serviceClass 接口类型,根据哪个接口生成子类代理对象;   providerParam :  "UserService#sayHello#"
    public static Object createProxy(Class<?> serviceClass) {
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
                new Class[]{serviceClass}, new InvocationHandler() {
                    @Override
                    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                        // 设置RpcRequest: UUID/接口全限定类名/方法名称/实参/入参类型
                        RpcRequest rpcRequest = new RpcRequest(UUID.randomUUID().toString(),
                                method.getDeclaringClass().getName(),
                                method.getName(),
                                args,
                                method.getParameterTypes());
                        /*if (userClientHandler == null) {
                            initClient(rpcRequest);
                        }*/
                        if (userClientHandlerMap == null) {
                            initClient(rpcRequest);
                        }
                        // 随机获取一个handler进行事件处理
//                        UserClientHandler userClientHandler = getUserClientHandlerRandom(rpcRequest);

                        // 获取响应时间最小的handler进行事件处理
                        UserClientHandler userClientHandler = getUserClientHandlerTime(rpcRequest);

                        userClientHandler.setRpcRequest(rpcRequest);
                        // 使用线程池,开启一个线程处理处理call() 写操作,并返回结果
                        Object rpcResponse = executorService.submit(userClientHandler).get();
                        return rpcResponse;
                    }
                });

    }


    // 随机获取一个handler进行事件处理
    public static UserClientHandler getUserClientHandlerRandom(RpcRequest rpcRequest) throws Exception {
        // 模拟随机获取所有在线服务器中的一个,随机key值获取handler
        Random random = new Random();
        List<String> keys = new ArrayList<>(userClientHandlerMap.keySet());
        String randomKey = keys.get(random.nextInt(keys.size()));
        return userClientHandlerMap.get(randomKey);
    }

    // 获取响应时间最小的handler进行事件处理
    public static UserClientHandler getUserClientHandlerTime(RpcRequest rpcRequest) throws Exception {
        /*String servicePath = "/"+rpcRequest.getInterfaceName();
        List<String> childrens = zkClient.getChildren().forPath(servicePath);
        childrens.forEach(server -> {

        });*/

        // 根据响应时间进行获取: 选出响应时间最小的server
        // 保存服务器所对应的响应时间
        Map<Long,UserClientHandler> timeMap = new TreeMap<>();
        for (String server : userClientHandlerMap.keySet()) {
            String serverPath = "/"+rpcRequest.getInterfaceName()+"/"+server;
            if (zkClient.checkExists().forPath(serverPath) == null) {
                continue;
            }
            UserClientHandler userClientHandler = userClientHandlerMap.get(server);
            Stat stat = new Stat();
            zkClient.getData().storingStatIn(stat).forPath(serverPath);
            long mtime = stat.getMtime();  //最后修改时间
            long responseTime = System.currentTimeMillis()-mtime;   //响应时间
            timeMap.put(responseTime, userClientHandler);   // server
            System.out.println(server+" 耗时===>>> "+responseTime);
            // 判断是否失效
            if (responseTime > 2200000) {   // 5000
                timeMap.put(0L, userClientHandler);    // 响应时间清零 同一台server
                System.out.println(server+" 在5s内没有新的请求,响应时间清零或失效");
            }
        }

        Long next = timeMap.keySet().iterator().next();
        return timeMap.get(next);
    }

}
